package com.gitee.dufafei.spark.conf

import org.apache.spark.SparkConf
import org.apache.spark.serializer.KryoSerializer

class SparkConfBuilder {

  private[this] val conf = new SparkConf()

  def setOption(key: String, value: String): this.type = {
    conf.set(key, value)
    this
  }

  def get(): SparkConf = {
    if (!conf.contains(SparkConfBuilder.SPARK_MASTER)) {
      conf.set(SparkConfBuilder.SPARK_MASTER, "local[2]")
    }
    conf
  }

  def setAppName(name: String): this.type = {
    setOption(SparkConfBuilder.SPARK_APP_NAME, name)
    this
  }

  def setKryoSerializer(): this.type = {
    setOption(SparkConfBuilder.SPARK_SERIALIZER, classOf[KryoSerializer].getName)
    this
  }

  def setSparkYarnJars(path: String): this.type = {
    setOption(SparkConfBuilder.SPARK_YARN_JARS, path)
    this
  }

  def setHiveMetaStoreUris(value: String): this.type = {
    setOption(SparkConfBuilder.SPARK_HIVE_URIS, value)
    this
  }

  def setDefaultParallelism(partitions: Int): this.type = {
    setOption(SparkConfBuilder.SPARK_DEFAULT_PARALLELISM, partitions.toString)
    this
  }

  def setSqlShufflePartitions(partitions: Int): this.type  = {
    setOption(SparkConfBuilder.SPARK_SQL_SHUFFLE_PARTITIONS, partitions.toString)
    this
  }

  // spark-streaming

  def setStreamingStopGracefullyOnShutdown(boolean: Boolean): this.type = {
    setOption(SparkConfBuilder.SPARK_STREAMING_STOP_GRACEFULLY_ON_SHUTDOWN, boolean.toString)
    this
  }

  def setStreamingBackpressureEnabled(boolean: Boolean): this.type = {
    setOption(SparkConfBuilder.SPARK_STREAMING_BACKPRESSURE_ENABLED, boolean.toString)
    this
  }

  def setStreamingKafkaMaxRatePerPartition(number: Int): this.type = {
    setOption(SparkConfBuilder.SPARK_STREAMING_KAFKA_MAX_RATE_PER_PARTITION, number.toString)
    this
  }
}

/**
 * 常用参数
 */
object SparkConfBuilder {

  def apply(): SparkConfBuilder = new SparkConfBuilder()

  // spark-core参数

  /**
   * 提交集群需要注释local：
   * 因为本地模式下，driver不会向resource manager申请资源，所以也就不会向rm注册自己。
   * 这也就解释了为什么在yarn上一直处于ACCEPTED状态，且应用日志中提示应用未向rm注册。
   */
  val SPARK_MASTER = "spark.master"
  val SPARK_APP_NAME = "spark.app.name"
  /**
   * 设置序列化方式,推荐kvro
   */
  val SPARK_SERIALIZER = "spark.serializer"
  /**
   * 指定应用的依赖jar在hdfs上的路径。
   * 示例：hdfs://hadoop001:9000/spark-yarn/jars/\*.jar
   * 替代：
   * --jars:
   * 在运行时把jar包分发到worker的指定目录上，一般是/var/run/spark/work目录，但是并不会把这些jar包自动装载到executor的classpath中
   *
   * spark.driver.extraClassPath：
   * spark.executor.extraClassPath：
   * 显式地将jars引入的jar包注册到executor的classpath中，因为executor知道运行的默认目录，所以不需要指定绝对目录，直接使用jar包名字即可
   * 在yarn-client模式下，–-jars 并不会把jar包拷贝到本机的目录下，所以在引用的时候要使用绝对路径
   */
  val SPARK_YARN_JARS = "spark.yarn.jars"
  /**
   * 全局参数
   * spark-core reduce阶段的task个数。
   * 一个作业一旦设置了该参数,它运行过程中的所有阶段的reduce个数都是同一个值。
   */
  val SPARK_DEFAULT_PARALLELISM = "spark.default.parallelism"

  // spark-sql参数

  /**
   * 如果要开启hive支持，必须设置此参数。
   */
  val SPARK_HIVE_URIS = "hive.metastore.uris"
  /**
   * 全局参数
   * spark-sql reduce阶段的task个数取(默认值200)。
   * 一个作业一旦设置了该参数，它运行过程中的所有阶段的reduce个数都是同一个值。
   */
  val SPARK_SQL_SHUFFLE_PARTITIONS = "spark.sql.shuffle.partition"

  // spark-streaming参数

  /**
   * 是否开启背压,即动态调节消费数据的速率，最大值不会超过KafkaMaxRatePerPartition设置的值。
   */
  val SPARK_STREAMING_BACKPRESSURE_ENABLED = "spark.streaming.backpressure.enabled"
  /**
   * 设置Kafka每个分区每秒能被消费的最大数量
   */
  val SPARK_STREAMING_KAFKA_MAX_RATE_PER_PARTITION = "spark.streaming.kafka.maxRatePerPartition"
  /**
   * 如果为true，Spark将在JVM关闭时优雅地关闭StreamingContext,而不是立即关闭。
   */
  val SPARK_STREAMING_STOP_GRACEFULLY_ON_SHUTDOWN = "spark.streaming.stopGracefullyOnShutdown"
}